/**
 * Copyright 2007-2008. Chongqing First Information & Network Co., Ltd. All
 * rights reserved. <a>http://www.cqfirst.com.cn</a>
 */
package com.cqcis.uip.base.framework.base.adaptor;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.StringTokenizer;
import java.util.WeakHashMap;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.mina.common.NoSerializable;
import org.osgi.framework.BundleContext;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.osgi.context.BundleContextAware;

import com.cqcis.uip.base.framework.FrameworkConstants;
import com.cqcis.uip.base.framework.UIPServiceException;
import com.cqcis.uip.base.framework.UIPTaskException;
import com.cqcis.uip.base.framework.common.Attributes;
import com.cqcis.uip.base.framework.common.HttpParam;
import com.cqcis.uip.base.framework.common.State;
import com.cqcis.uip.base.framework.common.Task;
import com.cqcis.uip.base.framework.common.TaskRecord;
import com.cqcis.uip.base.framework.common.Task.More;
import com.cqcis.uip.base.framework.common.Task.Source;
import com.cqcis.uip.base.framework.common.Task.Type;
import com.cqcis.uip.base.framework.common.impl.AttributesImpl;
import com.cqcis.uip.base.framework.core.AfterHandlerProcessResult;
import com.cqcis.uip.base.framework.core.Endpoint;
import com.cqcis.uip.base.framework.core.RegisteredService;
import com.cqcis.uip.base.framework.core.RegisteredServiceContext;
import com.cqcis.uip.base.framework.core.RegisteredServiceEvent;
import com.cqcis.uip.base.framework.core.RegisteredServiceEventType;
import com.cqcis.uip.base.framework.core.RegisteredServiceListener;
import com.cqcis.uip.base.framework.core.TaskWaitTimeConsumePool;
import com.cqcis.uip.base.framework.core.TaskWaitTimeoutExecutor;
import com.cqcis.uip.base.framework.core.ThreadPoolConfig;
import com.cqcis.uip.base.framework.core.channel.Channel;
import com.cqcis.uip.base.framework.core.cluster.ClusterSwitchSupport;
import com.cqcis.uip.base.framework.core.impl.TaskWaitTimeConsumePoolExecutorAdaptor;
import com.cqcis.uip.base.framework.core.impl.TaskWaitTimeConsumePoolImpl;
import com.cqcis.uip.base.framework.core.service.RegisteredServiceContextImpl;
import com.cqcis.uip.base.framework.service.StreamLogOutputService;
import com.cqcis.uip.base.framework.utils.ConcurrentTaskQueue;
import com.cqcis.uip.base.framework.utils.EndpointGenerator;
import com.cqcis.uip.base.framework.utils.LogServiceThreadPoolExecutor;
import com.cqcis.uip.base.framework.utils.LoopThreadPoolExecutor;
import com.cqcis.uip.base.framework.utils.StringUtil;
import com.cqcis.uip.base.framework.utils.TaskUtil;
import com.cqcis.uip.base.framework.utils.ConcurrentTaskQueue.EmptyBlockListener;
import com.cqcis.uip.base.framework.utils.ConcurrentTaskQueue.OfferedListener;
import com.cqcis.uip.base.logadaptor.LogConfiguration;

/**
 * 注册服务的抽象类, 任何注册服务的具体实现必须继承该类.
 * 
 * @author huangmy
 * @date   2008-1-10
 */
public abstract class AbstractRegisteredService implements RegisteredService, BundleContextAware, 
                          AfterHandlerProcessResult, NoSerializable, ApplicationContextAware {
	private static Log log = LogFactory.getLog(AbstractRegisteredService.class);
	private final static Task VIRTUAL_TASK = new Task(AbstractRegisteredService.class.getName());
	private final static String PARAM_CLUSTER_KEY = "PARAM_CLUSTER_KEY_" + AbstractRegisteredService.class.getName();
	private volatile boolean started;   // 已经启动标志, start() 方法调用后为true
	private volatile boolean stopped;   // 已经停止标志, stop() 方法调用后为true
	protected int waitTaskMax;               /*等待处理的最大任务数量*/
    private Attributes<Object> attributes = new AttributesImpl<Object>();   
    protected String endpointExpression; // 端点表达式
    protected String interfaceId;           /*接口ID, 接口程序的惟一标识*/
    protected String interfaceName;         /*接口名称*/
    private ConcurrentTaskQueue<Task>  taskQueue = new ConcurrentTaskQueue<Task>(); /*待处理任务队列*/
	private ConcurrentTaskQueue<Task>  taskResultQueue = new ConcurrentTaskQueue<Task>(); /*任务结果队列*/	
	private ConcurrentTaskQueue<Task> sendTaskResultQueue = new ConcurrentTaskQueue<Task>();// 发送任务结果队列
	private ConcurrentTaskQueue<Task> sendTaskResultControlQueue = new ConcurrentTaskQueue<Task>();// 发送任务结果控制队列
	private ConcurrentTaskQueue<Task> sendTaskResultPriorityQueue = new ConcurrentTaskQueue<Task>();// 发送任务结果优先级队列
	private ConcurrentTaskQueue<Task> sendTaskQueue = new ConcurrentTaskQueue<Task>();	// 发送任务队列	
	protected long intervalTime = 5000; /* 处理时间间隔 */	
	protected Endpoint toEndpoint; // 端点默认目的表达式
	protected int threadNumberTaskDispatcher = 1;  /*任务分发线程数量*/
	protected int threadNumberTaskResultDispatcher = 1;  /*任务结果分发线程数量*/
	protected int threadNumberExtendTaskResultProcessor = 1;  /*任务结果处理线程数量*/
	private transient LoopThreadPoolExecutor executor;    // 工作线程池
	protected transient ThreadPoolExecutor extendProcessResultThreadPoolExecutor; // 扩展结果处理线程池
	protected String logDir;   // 日志目录
	protected String logLevel; // 日志级别
	private transient LogConfiguration logConfiguration; // 日志配置
	protected transient BundleContext bundleContext; // OSGI bundle上下文
	private TaskRecord taskRecord = new TaskRecord(); // 任务处理记录
	private transient RegisteredServiceContext serviceContext; // 注册服务上下文
	private AtomicLong pendingSchedulesRepo;// 未决提交
	private transient ThreadPoolConfig threadPoolConfig = new ThreadPoolConfig(1, 12, 60); // 线程池配置
	private Object doScheduleLock = new Object(); // 提交同步
	private transient Map<String, Task> noResultTasks = new WeakHashMap<String, Task>() {
		@Override
		public int size() {
			int size = super.size();
			if (size <= 0) {
				return 0;
			}
			return size;
		}
	};
	private AtomicInteger pendingSendTaskSize = new AtomicInteger();
	protected StreamLogOutputService streamLogOutputService; // stream log
	private ClassLoader classLoader;
	private ApplicationContext applicationContext; // Application Context
	
	public void setApplicationContext(ApplicationContext applicationContext)
			throws BeansException {
		this.applicationContext = applicationContext;
	}
	
	/**
	 * 获取应用上下文
	 * 
	 * @return
	 */
	public ApplicationContext getApplicationContext() {
		return this.applicationContext;
	}
	
	public void registerClassLoader(ClassLoader classLoader) {
		this.classLoader = classLoader;
	}
	
	public ClassLoader getClassLoader() {
		return this.classLoader;
	}
			
	public Map<String, Task> getNoResultTasks() {		
		return this.noResultTasks;
	}
	
	public int getPendingSendTaskSize() {		
		return this.pendingSendTaskSize.get();
	}
	
	public void incrPendingSendTaskSize() {
		this.pendingSendTaskSize.incrementAndGet();
	}
	
	public void decrPendingSendTaskSize() {		
		if (this.pendingSendTaskSize.get() > 0) {
			this.pendingSendTaskSize.decrementAndGet();
		}		
	}
		
	/**
	 * 处理结果之后
	 * 
	 * @param result
	 */
	public void afterProcessResult(Task result) {
		String key = result.getToUnique();		
		Task task = this.noResultTasks.get(key);		
		if (task != null) {
			Object check = task.getAttribute(FrameworkConstants.TASK_IGNORE_AFTERPROCESSRESULT);
			if (check != null) {
				return;
			}
			// lock
			synchronized (task) {				
				this.noResultTasks.remove(key);
				ConcurrentTaskQueue<Task> relas = result.getRela();
				if (relas != null && relas.size() > 0) {
					Task rela = null;
					while ((rela = relas.remove()) != null) {
						getTaskQueue().offer(rela);
					}
				}
			}
		}		
	}
	
	/**
	 * 处理提交任务前其他操作
	 * @param task
	 * @return
	 */
	protected abstract boolean doSchedule();
	
	/**
	 * 同步，同 protected abstract boolean doSchedule(Task task);
	 * 
	 * @param task
	 * @return
	 */
	protected abstract boolean doScheduleSync(Task task);
	
	public ThreadPoolConfig getDefaultThreadPoolConfig() {		
		return this.threadPoolConfig;
	}
	
	public void catchException(Task task, Throwable t) {
		task.setState(State.ERROR);	
		String preMsg = (t instanceof UIPTaskException) ? "" : (FrameworkConstants.TASK_STATE_MSG_INTERFACE_PROCESS_ERROR + ":");
		task.setMessage(preMsg + t.getMessage(), t);
		getTaskResultQueue().offer(task);
	}
	
	/**
	 * 阻塞等待task处理完成, 适用于先等到taskResult，再等到finishTask的RS
	 * 
	 * @param task
	 * @return
	 */
	protected boolean waitForSyncTaskFinishBlock(Task task) {
		if (!waitForSyncTaskFinishBlockResult(task)) {			
			return false;
		}
		
		boolean s = waitForSyncTaskFinishBlockFinished(task);
		
		return s;
	}
	
	/**
	 * 阻塞等待同步taskResult
	 * 
	 * @param task
	 * @return
	 */
	protected boolean waitForSyncTaskFinishBlockResult(Task task) {
		Task processResult = TaskUtil.createOrGetSyncCTQFor(task, FrameworkConstants.TASK_SYNCCTQ_FOR_RESULT).poll();		
		if ((processResult == null) 
				|| (processResult != null && processResult.getState() == State.ERROR)) {
			doTaskResultRecordForRS(task);
			return false;
		}
		
		doTaskResultRecordForRS(processResult);
		
		return true;
	}
	
	/**
	 * 处理结果.
	 * 
	 * @param task
	 * @param exceptionResult 是否异常结果
	 */
	protected void dispatchTaskResult(Task task, boolean exceptionResult) {
		if (TaskUtil.isSyncTask(task)) {
			TaskUtil.createOrGetSyncCTQFor(task, FrameworkConstants.TASK_SYNCCTQ_FOR_RESULT).offer(task);
			if (exceptionResult) {
				TaskUtil.createOrGetSyncCTQFor(task, FrameworkConstants.TASK_SYNCCTQ_FOR_TASK).offer(task);
			}
		} else {
		    getTaskResultQueue().offer(task);	
		}
	}
	
	/**
	 * 阻塞等待同步taskFinished
	 * 
	 * @param task
	 * @return
	 */
	protected boolean waitForSyncTaskFinishBlockFinished(Task task) {
		Task result = TaskUtil.createOrGetSyncCTQFor(task, FrameworkConstants.TASK_SYNCCTQ_FOR_TASK).poll();				
		if ((result == null)
				|| (result != null && result.getState() == State.ERROR)) {
			doTaskResultRecordForRS(task);
			return false;
		}
		doTaskResultRecordForRS(result);
		return true;
	}
		
	/**
	 * task等待池
	 * 
	 * @return
	 */
	protected TaskWaitTimeConsumePool getTaskWaitTimeConsumePool() {
		final TaskWaitTimeConsumePoolImpl twtcp = new TaskWaitTimeConsumePoolImpl(new TaskWaitTimeoutExecutor() {
			public void execute(final Task task) {
				extendProcessResultThreadPoolExecutor.execute(new Runnable() {
					public void run() {
						try {
							scheduleSyncFor(task, false);
						} catch (Throwable t) {
							log.error("Catch Exception in schedule time wait task.", t);
						}
					}
				});				
			}
		});
		
		TaskWaitTimeConsumePool p = new TaskWaitTimeConsumePoolExecutorAdaptor(twtcp) {
            private boolean firstTimePut;
            private Object lock = new Object();
			
			@Override
			public synchronized void putTaskWait(Task task) { // 第一次调用put的时候启用线程监听
				synchronized (lock) {
					if (!firstTimePut) {
						firstTimePut = true;
						executor.execute(new Runnable() {
							public void run() {
								twtcp.run();
							}
						});
					}
				}
								
				super.putTaskWait(task);
			}
		};
				
		return p;
	}
	
	/**
	 * 获取该bundle的上下文
	 */
	public void setBundleContext(BundleContext context) {
	    this.bundleContext = context;
	}
	
	public BundleContext getBundleContext() {		
		return bundleContext;
	}
	
	public RegisteredServiceContext getRegisteredServiceContext() {		
		return serviceContext;
	}
		
	public LogConfiguration getLogConfiguration() {	
		if (logConfiguration == null) {
			logConfiguration = new LogConfiguration(logDir, logLevel);
		}
		
		return logConfiguration;
	}
	
	public void registerLogConfiguration(LogConfiguration logConfiguration) {
		this.logConfiguration = logConfiguration;
	}
	
	public TaskRecord getTaskRecord() {		
		return taskRecord;
	}
	
	public String getId() {		
		return getInterfaceId();
	}
	
	public String getName() {		
		return getInterfaceName();
	}
	
	@Override
	public String toString() {
		StringBuffer buffer = new StringBuffer();
		
		buffer.append("name: ").append(interfaceName)
		      .append(", id: ").append(interfaceId);
		
		return buffer.toString();
	}
	
	public boolean hasPendingSchedule() {		
		return pendingSchedulesRepo.get() > 0 ? true : false;
	}
	
	private synchronized boolean forSchedule(Task task) {
		if (stopped) {
			task.setMessage("RS has been stopped.");
			return false;
		}
		boolean f = decrFromRSPendingSendSize(task);
		if (!f) {			
			return true;
		}
		return scheduleFor(task, true);
	}
	
	public synchronized boolean schedule(Task task) {
		boolean s = forSchedule(task);
		if (!s) {
			if (task.getState() != State.ERROR) {
			    task.setState(State.ERROR);		
			}
		}
		return s;
	}
	
	public synchronized boolean scheduleCluster(Object obj) throws Exception {
		if (!(obj instanceof HttpParam)) {
			return false;
		}
		
		ClassLoader cl = getClassLoader();
						
		HttpParam p = (HttpParam)obj;		
		Task task = (Task)p.getContents().getObject(cl);		
		if (task == null) {
			throw new UIPServiceException("the task object can't be null.");
		}
		
		task.setSource(Source.REMOTE);
		task.setAttribute(PARAM_CLUSTER_KEY, p);
		if (!p.isOK()) {
			task.setState(State.ERROR);
		}
		
		if (task.getType() == Type.NORMAL) {
			return this.schedule(task);
		} else {
			return this.processResult(task);
		}
	}
	
	protected synchronized boolean scheduleFor(Task task, boolean record) {
		pendingSchedulesRepo.getAndIncrement();
		boolean s = false;
		try {
		    s = scheduleTaskForRS(task, record);
		} finally {
		    pendingSchedulesRepo.getAndDecrement();
		}
		return s;
	}
	
	private boolean scheduleTaskForRS(Task task, boolean record) {
		boolean success = acceptable() > 0 ? getTaskQueue().offer(task) : false;
		if (!success) {
			synchronized (task) {
				try {
					task.wait(10 * 1000);
					success = acceptable() > 0 ? getTaskQueue().offer(task) : false;					
				} catch (InterruptedException e) {
					return false;
				}
			}
			
			if (!success) {
			    task.setMessage("acceptable: " + acceptable());
			    return false;
			}
		}
			
		task.setMore(new More() {
			private static final long serialVersionUID = 1L;

			public boolean hasMore() {
				Task task = getTaskQueue().peek();
				return task == null ? false : true;
			}

		});
		
		if (record) {
			doTaskRecordForRS(task); // Task日志记录
		}
						
		synchronized (this.doScheduleLock) {
			boolean s = doSchedule();
			if (!s) {
			    task.setMessage("doSchedule Method return false");
			}
			return s;
		}		
	}
	
	/**
	 * Task日志记录
	 * 
	 * @param task
	 */
	private void doTaskRecordForRS(Task task) {
		taskRecord.getRecvTaskRepository().incrWait(); // 接收到的未处理数
		Endpoint fromEndpoint = task.route().from();
		RegisteredService rs = serviceContext.getServiceRepository()
                               .getRegisteredService(fromEndpoint); // 任务源服务
		if (rs != null) {
			rs.getTaskRecord().getSendTaskRepository().incrWait(); // 源服务发送的任务未处理数
		}
	}
	
	private boolean scheduleSyncFor(Task task, boolean record) {
		pendingSchedulesRepo.getAndIncrement();
		boolean s = false;
		try {
			if (record) {
				doTaskRecordForRS(task); // Task日志记录
			}
		    s = doScheduleSync(task);
		    if (!s && !StringUtil.hasText(task.getMessage())) {
		    	task.setMessage("doScheduleSync Method return false.");
		    }
		} finally {
		    pendingSchedulesRepo.getAndDecrement();
		}
		
		return s;
	}
	
	private boolean forScheduleSync(Task task) throws Exception {
		if (stopped) {
			task.setMessage("RS has been stopped.");
			return false;
		}
		boolean f = decrFromRSPendingSendSize(task);
		if (!f) {
			return true;
		}
		TaskUtil.setSyncTask(task);// 同步		
		boolean s = scheduleSyncFor(task, true);
		AbstractRegisteredService rs = (AbstractRegisteredService)serviceContext.getServiceRepository()
				.getRegisteredService(task.route().from()); // 任务源服务
		if (rs != null) {
			rs.afterProcessResult(task);
		}
		
		return s;
	}
	
	public boolean scheduleSync(Task task) throws Exception {
		boolean s = forScheduleSync(task);
		if (!s) {
			if (task.getState() != State.ERROR) {
			    task.setState(State.ERROR);		
			}
		}
		
		return s;
	}
	
	/**
	 * 修正计数
	 * 
	 * @param task
	 * @return true 继续 false 不再继续执行task
	 */
	private boolean decrFromRSPendingSendSize(Task task) {
		Endpoint from = task.route().from();
		AbstractRegisteredService fromRS = (AbstractRegisteredService)serviceContext
		                        .getServiceRepository().getRegisteredService(from);			
		if (fromRS != null) {						
			fromRS.decrPendingSendTaskSize();
		}
		
		Task r = this.noResultTasks.get(task.getFromUnique());
		if (r != null) {
			synchronized (r) {
				r = this.noResultTasks.get(task.getFromUnique());
				if (r != null && r != task) {
					r.addRela(task);
					return false;
				}
			}			
		}
		
		return true;
	}
	
	private void doStart(final RegisteredServiceContext serviceContext) throws Exception {
		int number = threadNumberTaskDispatcher + threadNumberTaskResultDispatcher + 3;
		if (number == 0) {
			number = 12;
		}
		
		this.serviceContext = serviceContext;
		
		executor = LoopThreadPoolExecutor.getExecutor(number, number * 2, 60,
				serviceContext.getLogService(), getLogConfiguration(), "(D)"+getInterfaceName(), getIntervalTime());
		
		for (int i = 0; i < threadNumberTaskDispatcher; i ++) {
			executor.execute(new Runnable() {
				public void run() {
					processSendTasks(serviceContext);
				}
			});
		}		
		
		for (int i = 0; i < threadNumberTaskResultDispatcher; i ++) {
			executor.execute(new Runnable() {
				public void run() {
					processTaskResults(serviceContext);
				}
			});
		}
		
		extendProcessResultThreadPoolExecutor = LogServiceThreadPoolExecutor.getLogExecutor(threadNumberExtendTaskResultProcessor, threadNumberExtendTaskResultProcessor * 2,
				60, serviceContext.getLogService(), getLogConfiguration(), "(E)"+getInterfaceName());
	}
	
	private void doStop(RegisteredServiceContext serviceContext) throws Exception {
		if (executor != null) {
			executor.shutdown();
		}
		
		if (extendProcessResultThreadPoolExecutor != null) {
			extendProcessResultThreadPoolExecutor.shutdown();
		}
		
		// 清理日志配置
		LogConfiguration logConfig = getLogServiceConfiguration(serviceContext);
		if (logConfig != null) {
			logConfig.destroy(); // 销毁
			serviceContext.getLogService().tryCloseLogAppender(logConfig);// 试图关闭			
		}
	}
	
	private LogConfiguration getLogServiceConfiguration(RegisteredServiceContext serviceContext) {
		RegisteredService rs = serviceContext.getRegisteredService();
		if (rs != null) {
			return rs.getLogConfiguration();
		}
		return null;
	}
	
	public void start(final RegisteredServiceContext serviceContext) throws Exception {
		if (!started) {
			if (serviceContext instanceof RegisteredServiceContextImpl) {
				((RegisteredServiceContextImpl)serviceContext).setStreamLogOutputService(this.streamLogOutputService);
			}
			
			pendingSchedulesRepo = new AtomicLong(0);
			try {
				doStart(serviceContext);
			} catch(Exception e) {
				throw e;
			}
			
			this.taskQueue.addEmptyBlockListener(new EmptyBlockListener() {
				private static final long serialVersionUID = 1L;

				public void blocking() {// 等待task					
					serviceContext.publishRegisteredServiceEvent(
							new RegisteredServiceEvent(RegisteredServiceEventType.WAITTASK, 
									serviceContext.getRegisteredService()));
				}
			});
			
			serviceContext.addServiceListener(new RegisteredServiceListener() {
				public void seviceChanged(RegisteredServiceEvent event) {
					if (event.getType() == RegisteredServiceEventType.STOPPED) {
						Collection<Task> tasks = noResultTasks.values();
						if (tasks != null) {
							Collection<Task> needRemoveTasks = new ArrayList<Task>();
							for (Task task : tasks) {
								RegisteredService rs = serviceContext.getServiceRepository()
								                  .getRegisteredService(task.route().to());
								if (rs != null && rs == event.getRegisteredService()) {
									needRemoveTasks.add(task);
								}
							}
							for (Task t : needRemoveTasks) {
								decrPendingSendTaskSize();
								noResultTasks.remove(t.getToUnique());
							}
						}
					}
				}
			});
			
			getSendTaskQueue().addOfferedListener(new OfferedListener() {
				public void offered(Object t) {
					Task task = (Task)t;
					if (task.getAttribute(FrameworkConstants.HAVA_SET_PENDRESULT) == null) {
						noResultTasks.put(task.getToUnique(), task);
						incrPendingSendTaskSize();
						task.setAttribute(
								FrameworkConstants.HAVA_SET_PENDRESULT,
								Boolean.TRUE);
					}
				}
			});
			
			this.sendTaskResultQueue.addOfferedListener(new OfferedListener() {
				public void offered(Object t) {
					sendTaskResultControlQueue.offer(VIRTUAL_TASK);
				}
			});
			
			this.sendTaskResultPriorityQueue.addOfferedListener(new OfferedListener() {
				public void offered(Object t) {
					sendTaskResultControlQueue.offer(VIRTUAL_TASK);
				}
			});
												
			started = true;
		}
	}
		
	public void stop(RegisteredServiceContext serviceContext) throws Exception {
		if (!stopped) {
			stopped = true;
			try {
				doStop(serviceContext);
			} catch(Exception e) {
				throw e;
			}
		}
	}
	
	public Task scheduleResult() {		
		return getTaskResultQueue().poll();
	}
	
	/**
	 * 处理发送task
	 * 
	 * @param serviceContext
	 */
	private void processSendTasks(RegisteredServiceContext serviceContext) {
		ConcurrentTaskQueue<Task> queue = getSendTaskQueue();
		Channel channel = serviceContext.getChannel();		
				
		while (true) {
			try {
				Task task = queue.poll();
				if (task == null) {
					break;
				}
								
				anaylyTaskEndpointRoute(task);
                
				if (task.getState() == State.ERROR || task.getState() == State.INVALID) {					
					task.setMessage(FrameworkConstants.TASK_STATE_MSG_DISPATCH_ERROR + ",State:" + task.getState());	
					task.setState(State.ERROR);
					/*
					 * and this allResults must write to this registered
					 * service
					 */
					decrPendingSendTaskSize();
					processResult(task);
				} else {					
				    channel.schedule(task);
				}
			} catch (Throwable t) {
				log.error("Catch Throwable in schedule Task.", t);
				
				throw new RuntimeException(t);
			}
		}
		
	}
		
	public boolean processResult(Task result) {				
		return getSendTaskResultQueue().offer(result);
	}
	
	/**
	 * Task日志记录
	 * 
	 * @param task
	 */
	protected RegisteredService doTaskResultRecordForRS(Task result) {
		Endpoint fromEndpoint = result.route().from();
				
		RegisteredService rs = serviceContext.getServiceRepository()
		        .getRegisteredService(fromEndpoint); // 任务源服务
		
		if (result.getState() == State.SUCCESS) {
			taskRecord.getRecvTaskRepository().incrSuccess(); // 接收到的任务处理成功
			if (rs != null) {
				rs.getTaskRecord().getSendTaskRepository().incrSuccess(); // 源服务发送的任务处理成功数
			}					
		} else if (result.getState() == State.ERROR) {
			taskRecord.getRecvTaskRepository().incrErrorService(); // 接收到任务服务处理失败
			if (rs != null) {
			    rs.getTaskRecord().getSendTaskRepository().incrErrorService();// 源服务发送的任务服务处理失败
			}
		}
		
		return rs;
	}
	
	/**
	 * 处理任务结果
	 * 
	 * @param serviceContext
	 */
	private void processTaskResults(RegisteredServiceContext serviceContext) {		
		while (true) {
			try {
				/* this results must send to the request registered service */
				Task result = scheduleResult();
				if (result == null) {
					break;
				}
				
				result.setType(Type.RESULT);
				
				RegisteredService rs = doTaskResultRecordForRS(result);
				if (result.getSource() == Source.REMOTE
						|| result.route().to().getAttribute(EndpointGenerator.ENDPOINT_EXPRESSION_LOCATION) != null) {
					// remote
					HttpParam p = (HttpParam)result.getAttribute(PARAM_CLUSTER_KEY);
					if (p != null) {
						HttpParam rp = ClusterSwitchSupport.getResponseParams(result, p);
						rp.getSession().write(rp);
					}
				} else if (rs != null) {					
					rs.processResult(result);
				}
			} catch (Throwable t) {
				log.error("Catch Throwable in process TaskResult.", t);				
				throw new UIPTaskException(t);
			}
		}
	}
	
	public int acceptable() {
		if (stopped) {
			return 0;
		}
		
		if (getWaitTaskMax() == 0) {
			return 0;
		}
		
		int size = getWaitTaskMax() - getTaskQueue().size();
		
		return size > 0 ? size : 0;
	}
	
	/**
	 * 设置任务路由标识
	 * @param task
	 * @throws Exception
	 */
	private void anaylyTaskEndpointRoute(Task task) throws Exception {
		if (task.route().from() == null) {
			Collection<Endpoint> endpoints = getEndpoints();
			
			task.route().from(endpoints.iterator().next());
		}
				
		if (task.route().to() == null && toEndpoint != null) {
			task.route().to(toEndpoint);
		}		
	}
	
	protected long getIntervalTime() {
		return intervalTime;
	}

	public void setIntervalTime(long intervalTime) {
		this.intervalTime = intervalTime;
	}
	
	public ConcurrentTaskQueue<Task> getTaskQueue() {
		return taskQueue;
	}

	public ConcurrentTaskQueue<Task> getTaskResultQueue() {
		return taskResultQueue;
	}

	protected String getInterfaceId() {
		return interfaceId;
	}

	public void setInterfaceId(String intfId) {
		this.interfaceId = intfId;
	}

	public void setInterfaceName(String interfaceName) {
		this.interfaceName = interfaceName;
	}

	protected String getInterfaceName() {
		return interfaceName;
	}
    
	public Attributes<Object> getAttributes() {		
		return attributes;
	}
	
    public Collection<Endpoint> getEndpoints() {
        StringTokenizer st = new StringTokenizer(endpointExpression, ",");
        List<Endpoint> endpoints = new ArrayList<Endpoint>();
		
		while(st.hasMoreTokens()) {
			String tacheStr = st.nextToken();
			tacheStr = tacheStr.trim();
			if (tacheStr != null && tacheStr.length() > 0) {				
				Endpoint endpoint = EndpointGenerator.generate(tacheStr);
				if (endpoint != null) {
					endpoints.add(endpoint);
				}
			}			
		}
		
		return endpoints;
	}

	public void setEndpointExpression(String endpointExpression) {
		this.endpointExpression = endpointExpression;
	}

	public void setToEndpointExpression(String toEndpointExpression) {
		this.toEndpoint = EndpointGenerator.generate(toEndpointExpression);
	}	
	
	public void setThreadNumberTaskDispatcher(int threadNumberTaskDispatcher) {
		this.threadNumberTaskDispatcher = threadNumberTaskDispatcher;
	}
	
	public void setWaitTaskMax(int waitTaskMax) {
		this.waitTaskMax = waitTaskMax;
	}
	
	public int getWaitTaskMax() {
		return this.waitTaskMax;
	}

	public ConcurrentTaskQueue<Task> getSendTaskResultQueue() {
		return sendTaskResultQueue;
	}

	public ConcurrentTaskQueue<Task> getSendTaskQueue() {
		return sendTaskQueue;
	}
    
	/**
	 * HANDLE processResult线程池
	 * 
	 * @return
	 */
	protected ThreadPoolExecutor getExtendProcessResultThreadPoolExecutor() {
		return extendProcessResultThreadPoolExecutor;
	}

	public void setThreadNumberTaskResultDispatcher(int threadNumberTaskResultDispatcher) {
		this.threadNumberTaskResultDispatcher = threadNumberTaskResultDispatcher;
	}

	public void setThreadNumberExtendTaskResultProcessor(int threadNumberExtendTaskResultProcessor) {		
		this.threadNumberExtendTaskResultProcessor = threadNumberExtendTaskResultProcessor;
	}

	public void setLogDir(String logDir) {
		this.logDir = logDir;
	}

	public void setLogLevel(String logLevel) {
		this.logLevel = logLevel;
	}

	public ConcurrentTaskQueue<Task> getSendTaskResultPriorityQueue() {
		return sendTaskResultPriorityQueue;
	}

	public ConcurrentTaskQueue<Task> getSendTaskResultControlQueue() {
		return sendTaskResultControlQueue;
	}
	
	public void setStreamLogOutputService(StreamLogOutputService streamLogOutputService) {
		this.streamLogOutputService = streamLogOutputService;
	}
}
